package org.myorg.datastraemapi.window;


import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.myorg.datastraemapi.KafkaDsProducer;
import org.myorg.datastraemapi.model.Student;

import java.util.Properties;

public class TimeWindowDemo {

	public static void main(String[] args) throws Exception {
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);  // 设定时间特性

		Properties props = new Properties();
		props.put("bootstrap.servers", "172.18.1.87:9092");
		props.put("group.id", "metric-group");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("auto.offset.reset", "latest");

		env.addSource(
				new FlinkKafkaConsumer011<>(
						KafkaDsProducer.topic,
						new SimpleStringSchema(), // String 序列化
						props
				)
		)
				.setParallelism(1)
				.map(string -> JSON.parseObject(string, Student.class))
				.keyBy(student -> student.age)
				.timeWindow(Time.seconds(1))
				.reduce((a, b) -> new Student(a.id + b.id, a.age + b.age))
				.addSink(new PrintSinkFunction<>());


		env.execute("add sink");
	}
}
